package gossip.v2.protocol;

import gossip.v2.GossipCore;
import gossip.v2.message.GossipMessage;
import gossip.v2.message.HeartbeatMessage;
import gossip.v2.message.serializer.GossipMessageSerializer;
import org.ricks.net.extension.protocol.FixedLengthBytesProtocol;
import org.ricks.net.transport.AioSession;

public class GossipProtocol extends FixedLengthBytesProtocol<GossipMessage> {

    private final GossipCore core;

    public GossipProtocol(GossipCore gossipCore) {
        this.core = gossipCore;
    }

    private final GossipMessageSerializer serializer = new GossipMessageSerializer();

    @Override
    protected GossipMessage decode(byte[] bytes, AioSession aioSession) {
        GossipMessage gossipMessage = serializer.deserialize(bytes);
        if(gossipMessage instanceof HeartbeatMessage heartbeatMessage && heartbeatMessage.getHeartbeatType() == HeartbeatMessage.HeartbeatType.RESPONSE) {
            processHeartbeat(heartbeatMessage);
        }
        return gossipMessage;
    }


    private void processHeartbeat(HeartbeatMessage msg) {
        if (msg.getSender() != null) {
            core.getClusterManager().onHeartbeat(msg.getSender(), msg.getTimestamp());
        }
    }
}
